package com.fsql.utils;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;

import java.util.Collection;
import java.util.concurrent.ExecutionException;

public class FlinkRestClientUtils {

    private static RestClusterClient restClient;

    public FlinkRestClientUtils(RestClusterClient rest) {
        restClient = rest;
    }

    /**
     * 取消job
     * @param jobId
     */
    public static void cancelJob(String jobId) {
        try {
            restClient.cancel(JobID.fromHexString(jobId)).get();
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 取消job，同时返回savePoint路径
     * @param jobId
     * @param savepointDirectory
     */
    public static String cancelWithSavepoint(String jobId, String savepointDirectory) {
        try {
            return String.valueOf(restClient.cancelWithSavepoint(JobID.fromHexString(jobId), savepointDirectory).get());
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
        return null;
    }

    /**
     * 停止任务，并返回savePoint路径
     * @param jobId
     *      jobId
     * @param advanceToEndOfTime
     *      是否提前结束任务
     * @param savepointDirectory
     *      快照保存点路径
     * @return
     */
    public static String stopWithSavepoint(String jobId, boolean advanceToEndOfTime, String savepointDirectory) {
        try {
            return String.valueOf(restClient.stopWithSavepoint(JobID.fromHexString(jobId), advanceToEndOfTime, savepointDirectory).get());
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
        return null;
    }



    /**
     * 获取任务的状态
     * @param jobId
     * @return
     */
    public static JobStatus getJobStatus(String jobId) {
        try {
            return (JobStatus) restClient.getJobStatus(JobID.fromHexString(jobId)).get();
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
        return null;
    }

    /**
     * 获取任务的详细数据
     * @param jobId
     * @return
     */
    public static JobDetailsInfo getJobDetails(String jobId) {
        try {
            return (JobDetailsInfo) restClient.getJobDetails(JobID.fromHexString(jobId)).get();
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
        return null;
    }

    /**
     * 获取所有任务的状态
     * @return
     */
    public static Collection<JobStatusMessage> listJobs() {
        try {
            return (Collection<JobStatusMessage>) restClient.listJobs().get();
        } catch (ExecutionException | InterruptedException ex) {
            ex.printStackTrace();
        }
        return null;
    }

    /**
     * 获取web接口地址
     * @return
     */
    public static String getInterfaceUrl() {
        return restClient.getWebInterfaceURL();
    }
}
